package com.we.risk.smsInfo.HiveDataBackup;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.we.flink.utils.WeKafkaPropertyReader;
import com.we.risk.smsInfo.SmsCategory;
import com.we.risk.smsInfo.SmsInfoBasic;
import com.we.utils.CommonTimeTools;
import com.we.utils.ParamCheck;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.text.SimpleDateFormat;
import java.util.*;

import static com.we.risk.smsInfo.SmsType.*;

public class AdmSmsInfoLabelFromHiveTest {
    public static final String RELEASEPROP = "risk/smsInfo/adm_sms_hive_backup.properties";
    public static final String PROPSETS = "risk/smsInfo/sms230407_reloan.properties";
    public static Logger LOG = LoggerFactory.getLogger(AdmSmsInfoLabelFromHiveTest.class);

    public static final String SPLITWORD = "#";
    public static final String DATEFORMAT = "yyyy-MM-dd";
    public static final String MSGTYPE = "HFQ_UPLOAD_SMS_REPORT_RECORD";
    public static final String INPUTKEYBY = "userKey";
    public static final String OUTPUTKEY = "user_key";
    public static final String HTTPTIMESTAMP = "httpTimestamp";
    public static final String CAPTIME = "capture_time";
    public static final String LABELOUTPUTIME = "label_output_time";
    public static final String INJOBID = "jobID";
    public static final String OUTJOBID = "job_id";
    public static final String SMSLIST = "sms_list";
    public static final int JSONSIZE = 4;

    public static void main(String[] args) throws IOException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        TableConfig tableConf = tableEnv.getConfig();
        Configuration selfConf = new Configuration();
        selfConf.setBoolean("table.exec.hive.infer-source-parallelism", false);
        tableConf.addConfiguration(selfConf);
        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** hive catlog configure */
            String hiveCatlogName = paramReader.getHiveCatlogName();
            String hiveDBName = paramReader.getHiveDBName();
            String hiveConfDir = paramReader.getHiveConfDir();

            HiveCatalog hiveCatalog = new HiveCatalog(hiveCatlogName, hiveDBName, hiveConfDir);
            tableEnv.registerCatalog("hive", hiveCatalog);
            tableEnv.useCatalog("hive");

            Table table = tableEnv.sqlQuery(TableSms.QUERYSQL);
            DataStream<TableSms> input = tableEnv.toDataStream(table, TableSms.class);

            SingleOutputStreamOperator<String> out =
                    input.map(
                                    new RichMapFunction<TableSms, String>() {
                                        @Override
                                        public String map(TableSms value) throws Exception {
                                            return value.getContent();
                                        }
                                    })
                            .flatMap(new smsInfoRichFlatMapFunction())
                            .keyBy(
                                    new KeySelector<String, String>() {
                                        @Override
                                        public String getKey(String value) throws Exception {
                                            JSONObject inputJson = JSONObject.parseObject(value);
                                            return inputJson.getString(OUTPUTKEY);
                                        }
                                    })
                            .flatMap(
                                    new RichFlatMapFunction<String, String>() {
                                        ValueState<JSONObject> plistVState;

                                        @Override
                                        public void open(Configuration parameters)
                                                throws Exception {
                                            plistVState =
                                                    getRuntimeContext()
                                                            .getState(
                                                                    new ValueStateDescriptor<
                                                                            JSONObject>(
                                                                            "smsVstate",
                                                                            JSONObject.class));
                                        }

                                        @Override
                                        public void flatMap(String value, Collector<String> out)
                                                throws Exception {
                                            try {
                                                JSONObject inputJson =
                                                        JSONObject.parseObject(value);
                                                JSONObject tmpState = plistVState.value();
                                                JSONObject curState =
                                                        tmpState == null
                                                                ? initPropJson(PROPSETS)
                                                                : tmpState;
                                                String outKey = inputJson.getString(OUTPUTKEY);
                                                Long input_cap_time = inputJson.getLong(CAPTIME);
                                                Long state_cap_time = curState.getLong(CAPTIME);

                                                if (input_cap_time > state_cap_time) {
                                                    Iterator<String> iterator =
                                                            curState.keySet().iterator();
                                                    while (iterator.hasNext()) {
                                                        String key = iterator.next();
                                                        if (!key.equals(OUTPUTKEY)
                                                                && !key.equals(CAPTIME)) {
                                                            Object labelV = inputJson.get(key);
                                                            Object state_v = curState.get(key);
                                                            if (state_v != null && labelV != null) {
                                                                if (!state_v.equals(labelV)) {
                                                                    JSONObject outJson =
                                                                            new JSONObject();
                                                                    outJson.put(OUTPUTKEY, outKey);
                                                                    outJson.put(
                                                                            CAPTIME,
                                                                            input_cap_time);
                                                                    outJson.put("label_name", key);
                                                                    outJson.put(
                                                                            "label_value", labelV);
                                                                    out.collect(outJson.toString());
                                                                    curState.put(key, labelV);
                                                                }
                                                            } else {
                                                                if (state_v != labelV) {
                                                                    JSONObject outJson =
                                                                            new JSONObject();
                                                                    outJson.put(OUTPUTKEY, outKey);
                                                                    outJson.put(
                                                                            CAPTIME,
                                                                            input_cap_time);
                                                                    outJson.put("label_name", key);
                                                                    outJson.put(
                                                                            "label_value", labelV);
                                                                    out.collect(
                                                                            outJson.toString(
                                                                                    SerializerFeature
                                                                                            .WriteMapNullValue));
                                                                    curState.put(key, labelV);
                                                                }
                                                            }
                                                        }
                                                    }
                                                    curState.put(CAPTIME, input_cap_time);
                                                    plistVState.update(curState);
                                                }
                                            } catch (Exception e) {
                                                LOG.error("FlatMap Exception: " + e.toString());
                                                LOG.error("FlatMap error input: " + value);
                                            }
                                        }
                                    });

            /** sink to Kafka */
            String sinkkfkTopic = paramReader.getKfkTopic();
            Properties sinkProp = new Properties();
            sinkProp.setProperty("bootstrap.servers", paramReader.getKfkBootStrapServer());
            //            sinkProp.setProperty("max.request.size", String.valueOf(KFKMSGMAXSIZE));
            int sinkkfkPartitions = paramReader.getKfkPartitions();

            FlinkKafkaProducer<String> kafkaProducer =
                    new FlinkKafkaProducer<String>(
                            sinkkfkTopic,
                            new WeKafkaKeyedSerializationSchema(),
                            sinkProp,
                            Optional.of(new WeKafkaCustomPartitioner()));

            out.addSink(kafkaProducer).setParallelism(sinkkfkPartitions);

            env.execute(AdmSmsInfoLabelFromHiveTest.class.toString());
        } catch (Exception e) {
            System.out.println("Exception: " + e.getMessage());
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static class smsInfoRichFlatMapFunction extends RichFlatMapFunction<String, String> {
        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            try {
                SimpleDateFormat sdf = new SimpleDateFormat(DATEFORMAT);
                JSONObject inputJson = filterSmsInfoMsg(value);
                if (inputJson.size() == JSONSIZE
                        && inputJson.containsKey(OUTPUTKEY)
                        && inputJson.containsKey(SMSLIST)) {
                    HashMap<String, SmsCategory> d7Map = initODMap();
                    HashMap<String, SmsCategory> d30Map = initMap();
                    HashMap<String, SmsCategory> d60Map = initODMap();
                    HashMap<String, SmsCategory> d90Map = initMap();
                    HashMap<String, SmsCategory> d180Map = initMap();
                    HashMap<String, SmsCategory> d365Map = initMap();

                    String userKey = inputJson.getString(OUTPUTKEY);
                    String jobId = inputJson.getString(OUTJOBID);
                    Long capTime = inputJson.getLong(CAPTIME);
                    JSONArray smsArray = inputJson.getJSONArray(SMSLIST);
                    int size = smsArray.size();
                    for (int i = 0; i < size; i++) {
                        JSONObject smsInfo = (JSONObject) smsArray.get(i);
                        ParamCheck.isNull(smsInfo, "smsInfo");

                        SmsInfoBasic smsInfoBasic =
                                SmsInfoBasic.constructSmsinfoBasic(smsInfo, capTime, sdf);
                        if (smsInfoBasic != null) {
                            long recvTime = smsInfoBasic.getRecvTime();
                            Integer dateDiff = CommonTimeTools.dateDiff(capTime, recvTime, sdf);

                            if (dateDiff <= 7) {
                                smsLoanOdProcess(smsInfoBasic, d7Map, sdf, dateDiff);
                            }

                            if (dateDiff <= 60) {
                                smsLoanOdProcess(smsInfoBasic, d60Map, sdf, dateDiff);
                            }

                            if (dateDiff <= 30) {
                                smsCateLabelProcess(smsInfoBasic, d30Map, sdf, dateDiff);
                            }
                            if (dateDiff <= 90) {
                                smsCateLabelProcess(smsInfoBasic, d90Map, sdf, dateDiff);
                            }
                            if (dateDiff <= 180) {
                                smsCateLabelProcess(smsInfoBasic, d180Map, sdf, dateDiff);
                            }
                            if (dateDiff <= 365) {
                                smsCateLabelProcess(smsInfoBasic, d365Map, sdf, dateDiff);
                            }
                        }
                    }
                    JSONObject outJson = new JSONObject();
                    outJson.put(OUTPUTKEY, userKey);
                    outJson.put(OUTJOBID, jobId);
                    outJson.put(CAPTIME, capTime);
                    collectLabelJsonObject(d7Map, outJson, "d7");
                    collectLabelJsonObject(d30Map, outJson, "d30");
                    collectLabelJsonObject(d60Map, outJson, "d60");
                    collectLabelJsonObject(d90Map, outJson, "d90");
                    collectLabelJsonObject(d180Map, outJson, "d180");
                    collectLabelJsonObject(d365Map, outJson, "d365");
                    long labelOutPutTime = System.currentTimeMillis();
                    outJson.put(LABELOUTPUTIME, labelOutPutTime);
                    out.collect(outJson.toString());
                }
            } catch (Exception e) {
                LOG.error(e.getMessage());
                LOG.error(e.toString());
                LOG.error("smsInfoRichFlatMapFunction Exception input: " + value);
            }
        }

        private void collectLabelJsonObject(
                HashMap<String, SmsCategory> map, JSONObject inputJson, String diffDays) {
            if (!map.isEmpty() && inputJson != null) {
                Set<String> keySet = map.keySet();
                for (String key : keySet) {
                    SmsCategory smsCategory = map.get(key);
                    if (smsCategory != null) {
                        smsCategory.smsCalcLable(diffDays);
                        HashMap<String, Object> resMap = smsCategory.getResMap();
                        inputJson.putAll(resMap);
                    }
                }
            }
        }

        private HashMap<String, SmsCategory> initMap() {
            HashMap<String, SmsCategory> map = new HashMap<>();

            map.put(BASE, new SmsCategory(BASE));
            map.put(TELE, new SmsCategory(TELE));
            map.put(EXPS, new SmsCategory(EXPS));
            map.put(SERV, new SmsCategory(SERV));
            map.put(PAYT, new SmsCategory(PAYT));
            map.put(BANK, new SmsCategory(BANK));
            map.put(CRDT, new SmsCategory(CRDT));
            map.put(LOAN, new SmsCategory(LOAN));
            map.put(OD, new SmsCategory(OD));

            return map;
        }

        private HashMap<String, SmsCategory> initODMap() {
            HashMap<String, SmsCategory> map = new HashMap<>();
            map.put(OD, new SmsCategory(OD));

            return map;
        }

        private void smsCateLabelProcess(
                SmsInfoBasic basicInfo,
                HashMap<String, SmsCategory> cateMap,
                SimpleDateFormat sdf,
                Integer diffDays) {
            try {
                cateMap.get(BASE).smsBaseProcess(basicInfo, sdf, diffDays);
                int isRecv = basicInfo.getIsRecv();
                String body_name = basicInfo.getBody_name();

                if (isRecv == 1 && body_name != null && body_name.length() >= 2) {
                    if (basicInfo.getIsTelecom() == 1 && cateMap.get(TELE) != null) {
                        cateMap.get(TELE).smsCateProcess(basicInfo, sdf);
                    }

                    if (basicInfo.getIsExpress() == 1 && cateMap.get(EXPS) != null) {
                        cateMap.get(EXPS).smsCateProcess(basicInfo, sdf);
                    }

                    if (basicInfo.getIsService() == 1 && cateMap.get(SERV) != null) {
                        cateMap.get(SERV).smsCateProcess(basicInfo, sdf);
                    }

                    if (basicInfo.getIsPayTool() == 1 && cateMap.get(PAYT) != null) {
                        cateMap.get(PAYT).smsCateProcess(basicInfo, sdf);
                    }

                    if (basicInfo.getIsBank() == 1 && cateMap.get(BANK) != null) {
                        cateMap.get(BANK).smsBankProcess(basicInfo, sdf);
                    }

                    if (basicInfo.getIsCrdt() == 1 && cateMap.get(CRDT) != null) {
                        cateMap.get(CRDT).smsCrdtProcess(basicInfo, sdf);
                    }

                    if (basicInfo.getIsItntLend() == 1 && cateMap.get(LOAN) != null) {
                        cateMap.get(LOAN).smsItntProcess(basicInfo, sdf);
                    }

                    if (basicInfo.getIsOd() == 1 && cateMap.get(OD) != null) {
                        cateMap.get(OD).smsOdProcess(basicInfo, sdf);
                    }
                }

            } catch (Exception e) {
                LOG.error("smsCateLabelProcess excep: " + e.toString());
            }
        }

        private void smsLoanOdProcess(
                SmsInfoBasic basicInfo,
                HashMap<String, SmsCategory> cateMap,
                SimpleDateFormat sdf,
                Integer diffDatys) {
            try {
                int isRecv = basicInfo.getIsRecv();
                String body_name = basicInfo.getBody_name();

                if (isRecv == 1 && body_name != null && body_name.length() >= 2) {
                    if (basicInfo.getIsOd() == 1 && cateMap.get(OD) != null) {
                        cateMap.get(OD).smsOdProcess(basicInfo, sdf);
                    }
                }

            } catch (Exception e) {
                LOG.error("smsLoanOdProcess excep: " + e.toString());
                LOG.error("smsLoanOdProcess input: " + basicInfo.toString());
            }
        }
    }

    private static JSONObject filterSmsInfoMsg(String value) {
        JSONObject res = new JSONObject();
        SimpleDateFormat sdf = new SimpleDateFormat(DATEFORMAT);
        try {
            if (value != null && value.contains(MSGTYPE)) {
                JSONObject inputJson = JSONObject.parseObject(value);
                if (inputJson != null) {
                    String type = inputJson.getString("type");
                    if (type != null && type.equals(MSGTYPE)) {
                        JSONObject datajson = inputJson.getJSONObject("dataJson");
                        if (datajson != null) {
                            JSONObject job = datajson.getJSONObject("job");
                            if (job != null) {
                                String job_id = job.getString(INJOBID);
                                if (job_id != null) {
                                    res.put(OUTJOBID, job_id);
                                }
                            }
                            String user_key = datajson.getString(INPUTKEYBY);
                            if (user_key != null) {
                                res.put(OUTPUTKEY, user_key);
                            }
                            JSONObject data = datajson.getJSONObject("data");
                            if (data != null) {
                                JSONObject baseData = data.getJSONObject("baseData");
                                if (baseData != null) {
                                    Long capture_time = baseData.getLong(HTTPTIMESTAMP);
                                    if (capture_time != null) {
                                        res.put(CAPTIME, capture_time);
                                        JSONArray smsInfoList = data.getJSONArray("smsInfoList");
                                        if (smsInfoList != null) {
                                            //                                            JSONArray
                                            // outJsonArray = jsonListFilter(smsInfoList,
                                            // capture_time, sdf);
                                            //                                            if
                                            // (outJsonArray.size() > 0) {
                                            //
                                            // res.put(SMSLIST, outJsonArray);
                                            //                                            }
                                            res.put(SMSLIST, smsInfoList);
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
        } catch (Exception e) {
            LOG.error(e.toString());
            LOG.error("filterSmsInfoMsg excep input: " + value);
        }
        return res;
    }

    private static JSONArray jsonListFilter(JSONArray input, Long cap_time, SimpleDateFormat sdf) {
        JSONArray res = new JSONArray();
        try {
            int size = input.size();
            for (int i = 0; i < size; i++) {
                JSONObject smsInfo = (JSONObject) input.get(i);
                Integer type = smsInfo.getInteger("type");
                Long date = smsInfo.getLong("date");
                if ((type != null && (type == 1 || type == 2))
                        && (date != null && CommonTimeTools.dateDiff(cap_time, date, sdf) <= 365)) {
                    res.add(smsInfo);
                }
            }

        } catch (Exception e) {
            LOG.error(e.toString());
            LOG.error("jsonListFilter excep input: " + input);
        }
        return res;
    }

    private static JSONObject initPropJson(String propFile) throws IOException {
        InputStream resourceAsStream =
                AdmSmsInfoLabelFromHiveTest.class.getClassLoader().getResourceAsStream(propFile);
        ParameterTool reader = ParameterTool.fromPropertiesFile(resourceAsStream);
        JSONObject res = new JSONObject();
        Iterator<Object> keysIterator = reader.getProperties().keySet().iterator();
        while (keysIterator.hasNext()) {
            String key = (String) keysIterator.next();
            if (key.equals(CAPTIME)) {
                res.put(key, -1);
            } else {
                res.put(key, "initV");
            }
        }
        return res;
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String keyby = jsonObject.getString(OUTPUTKEY);
            return keyby.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.info(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }
}
